package taillog

import (
	"Study_go/LogAgent/kafka"
	"context"
	"fmt"
	"github.com/hpcloud/tail"
)

var (
	tails *tail.Tail

)

type TailTask struct{
	Path string
	Topic string
	Tailobj *tail.Tail
	ctx context.Context
	cancelFunc context.CancelFunc
}

func NewTailTask(path string,topic string) (tailTask *TailTask){
	ctx,cancel:=context.WithCancel(context.Background())
	tailTask =&TailTask{
		Path:path,
		Topic:topic,
		ctx:ctx,
		cancelFunc:cancel,
	}
	tailTask.init()
	return

}

func (this *TailTask) init(){
	var err error
	this.Tailobj,err=tail.TailFile(this.Path,tail.Config{
		ReOpen:true,
		Follow:true,
		Location:&tail.SeekInfo{Offset:0,Whence:2},
		MustExist:false,
		Poll:true,
	})
	if err!=nil{
		fmt.Println("初始化配置文件出错")
		return
	}
	go this.SendToKafka()

}

func (this *TailTask)  SendToKafka(){
	for{
		select {
			case <-this.ctx.Done():
				fmt.Println("任务 %s_%s结束了",this.Path,this.Topic)
				return
			case line:=<- this.Tailobj.Lines:
				//kafka.SendMsg(this.Topic,line.Text) //这里存在函数调函数，所以改成向一个chan中发送数据
				kafka.SendToChan(this.Topic,line.Text)
		}
	}
}
//func Init(filename string)(err error){
//	tails,err=tail.TailFile(filename,tail.Config{
//		ReOpen:true,
//		Follow:true,
//		Location:&tail.SeekInfo{Offset:0,Whence:2},
//		MustExist:false,
//		Poll:true,
//	})
//	if err!=nil{
//		fmt.Println("初始化配置文件出错")
//		return
//	}
//	return
//}

func ReadData() <-chan *tail.Line{
	return tails.Lines
}